package org.apache.rocketmq.proxy.grpc.v2;

import apache.rocketmq.v2.*;
import io.grpc.Context;
import io.grpc.Metadata;
import io.grpc.stub.StreamObserver;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.thread.ThreadPoolMonitor;
import org.apache.rocketmq.common.utils.StartAndShutdown;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.config.ProxyConfig;
import org.apache.rocketmq.proxy.grpc.interceptor.InterceptorConstants;
import org.apache.rocketmq.proxy.grpc.v2.common.GrpcProxyException;
import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder;
import org.apache.rocketmq.proxy.grpc.v2.common.ResponseWriter;
import org.apache.rocketmq.proxy.processor.MessagingProcessor;
import org.apache.rocketmq.proxy.processor.channel.ChannelProtocolType;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;


/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/1/3
 * @方法描述：Grpc服务端消息处理应用，当Grpc客户端每一次执行远程调用方法的时候，服务端的这个类的对象，都会根据客户端远程调用的方法的名称，执行对应的方法
 * 这个类的对象就相当于服务端提供目标方法的服务组件
 */
public class GrpcMessagingApplication extends MessagingServiceGrpc.MessagingServiceImplBase implements StartAndShutdown {


    private final static Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);

    //Grpc服务端的默认消息处理器
    private final GrpcMessingActivity grpcMessingActivity;

    //Grpc服务端处理客户端发送过来的根据主题查询路由信息的请求时，就会使用这个线程池来处理
    protected ThreadPoolExecutor routeThreadPoolExecutor;

    //Grpc服务端处理客户端发送过来的客户端信息请求时，就会使用这个线程池来处理
    protected ThreadPoolExecutor clientManagerThreadPoolExecutor;

    //和生产者客户端相关的操作都由这个执行器执行
    protected ThreadPoolExecutor producerThreadPoolExecutor;

    //构造方法
    protected GrpcMessagingApplication(GrpcMessingActivity grpcMessingActivity) {
        this.grpcMessingActivity = grpcMessingActivity;
        //从配置管理器中得到Proxy模块的配置信息对象
        ProxyConfig config = ConfigurationManager.getProxyConfig();
        //创建routeThreadPoolExecutor线程池，用于处理主题路由信息请求
        this.routeThreadPoolExecutor = ThreadPoolMonitor.createAndMonitor(
                //得到主题路由信息线程池的核心线程数
                config.getGrpcRouteThreadPoolNums(),
                //得到主题路由信息线程池的最大线程数
                config.getGrpcRouteThreadPoolNums(),
                //得到主题路由信息线程池的线程空闲存活时间
                1,
                TimeUnit.MINUTES,
                "GrpcRouteThreadPool",
                //得到主题路由信息线程池的队列容量
                config.getGrpcRouteThreadQueueCapacity()
        );
        this.producerThreadPoolExecutor = ThreadPoolMonitor.createAndMonitor(
                config.getGrpcProducerThreadPoolNums(),
                config.getGrpcProducerThreadPoolNums(),
                1,
                TimeUnit.MINUTES,
                "GrpcProducerThreadPool",
                config.getGrpcProducerThreadQueueCapacity()
        );

        //创建clientManagerThreadPoolExecutor线程池，用于处理客戶端信息请求
        this.clientManagerThreadPoolExecutor = ThreadPoolMonitor.createAndMonitor(
                config.getGrpcClientManagerThreadPoolNums(),
                config.getGrpcClientManagerThreadPoolNums(),
                1,
                TimeUnit.MINUTES,
                "GrpcClientManagerThreadPool",
                config.getGrpcClientManagerThreadQueueCapacity()
        );


        //执行初始化方法
        this.init();
    }



    //该初始化方法的执行的操作非常简单，就是把线程池的拒绝策略处理器设置到线程池中
    protected void init() {
        GrpcTaskRejectedExecutionHandler rejectedExecutionHandler = new GrpcTaskRejectedExecutionHandler();
        this.routeThreadPoolExecutor.setRejectedExecutionHandler(rejectedExecutionHandler);
        this.clientManagerThreadPoolExecutor.setRejectedExecutionHandler(rejectedExecutionHandler);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/3
     * @方法描述：创建GrpcMessagingApplication对象的方法，在创建GrpcMessagingApplication对象的时候，也把其内部使用的默认消息处理器创建出来了
     */
    public static GrpcMessagingApplication create(MessagingProcessor messagingProcessor) {
        return new GrpcMessagingApplication(new DefaultGrpcMessingActivity(messagingProcessor));
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/3
     * @方法描述：生成响应状态码的方法，并且这里生成的是控制客户端请求流量的状态码，也就是告诉客户端请求发送得太频繁了，需要限流
     */
    protected Status flowLimitStatus() {
        //使用ResponseBuilder构建一个状态对象，状态码是TOO_MANY_REQUESTS，状态信息是flow limit
        return ResponseBuilder.getInstance().buildStatus(Code.TOO_MANY_REQUESTS, "flow limit");
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/3
     * @方法描述：将异常转换为Status状态对象的方法
     */
    protected Status convertExceptionToStatus(Throwable t) {
        return ResponseBuilder.getInstance().buildStatus(t);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/3
     * @方法描述：Grpc服务端提供的根据主题查询路由信息的目标方法，当Grpc客户端远程调用这个方法的时候，会向服务端发送一个QueryRouteRequest请求对象
     * 而请求对象就会被服务端的这个方法处理，处理完成后，会向客户端返回一个QueryRouteResponse响应对象
     */
    @Override
    public void queryRoute(QueryRouteRequest request, StreamObserver<QueryRouteResponse> responseObserver) {
        System.out.println("queryRoute操作执行了！================================================================================================");
        //首先定义一个函数，这个函数会根据传输的status状态对象生成一个QueryRouteResponse响应对象
        Function<Status, QueryRouteResponse> statusResponseCreator = status -> QueryRouteResponse.newBuilder().setStatus(status).build();
        //创建请求上下文信息对象
        ProxyContext context = createContext();
        try {
            //对刚才创建的上下文对象进行校验
            validateContext(context);
            //把服务端处理请求的操作封装成一个任务，提交给专门处理主题路由信息的线程池执行
            this.addExecutor(this.routeThreadPoolExecutor,
                    //传入请求上下文信息对象
                    context,
                    //传入请求对象
                    request,
                    //创建一个Runnable对象，这个Runnable对象封装的就是处理请求的操作
                    //可以看到这里直接调用了默认消息处理器的queryRoute方法处理请求
                    () -> grpcMessingActivity.queryRoute(context, request)
                            //在请求处理完毕之后会执行这里的操作，也就是回复响应给客户端
                            .whenComplete((response, throwable) ->
                                    //这里把responseObserver对象传入，是因为向客户端回复响应的就是这个对象
                                    writeResponse(context, request, response, responseObserver, throwable, statusResponseCreator)),
                    responseObserver,
                    statusResponseCreator);
        } catch (Throwable t) {
            //出现异常则直接回复错误响应
            writeResponse(context, request, null, responseObserver, t, statusResponseCreator);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：处理客户端心跳请求的方法，该方法逻辑和queryRoute方法类似，都是将处理逻辑封装成一个任务，提交给对应的线程池执行
     */
    @Override
    public void heartbeat(HeartbeatRequest request, StreamObserver<HeartbeatResponse> responseObserver) {

        System.out.println("heartbeat操作执行了！================================================================================================");

        Function<Status, HeartbeatResponse> statusResponseCreator = status -> HeartbeatResponse.newBuilder().setStatus(status).build();
        ProxyContext context = createContext();
        try {
            validateContext(context);
            this.addExecutor(this.clientManagerThreadPoolExecutor,
                    context,
                    request,
                    //在这里调用Grpc默认消息处理器的heartbeat方法处理心跳请求
                    () -> grpcMessingActivity.heartbeat(context, request)
                            .whenComplete((response, throwable) ->
                                    writeResponse(context, request, response, responseObserver, throwable, statusResponseCreator)),
                    responseObserver,
                    statusResponseCreator);
        } catch (Throwable t) {
            writeResponse(context, request, null, responseObserver, t, statusResponseCreator);
        }
    }


    //以下方法在第十版本代码中还没用到，所以我就先不实现了，之所以引入进来是应为第九版本代码的客户端已经存在对应方法了
    //所以现在先给服务端引入进来，大家也知道客户端发送的请求会被服务端的哪个方法处理

    @Override
    public void sendMessage(SendMessageRequest request, StreamObserver<SendMessageResponse> responseObserver) {
        Function<Status, SendMessageResponse> statusResponseCreator = status -> SendMessageResponse.newBuilder().setStatus(status).build();
        ProxyContext context = createContext();
        try {
            validateContext(context);
            this.addExecutor(this.producerThreadPoolExecutor,
                    context,
                    request,
                    () -> grpcMessingActivity.sendMessage(context, request)
                            .whenComplete((response, throwable) ->
                                    writeResponse(context, request, response, responseObserver, throwable, statusResponseCreator)),
                    responseObserver,
                    statusResponseCreator);

        }catch (Throwable t){
            writeResponse(context, request, null, responseObserver, t, statusResponseCreator);
        }
    }


    @Override
    public void queryAssignment(QueryAssignmentRequest request, StreamObserver<QueryAssignmentResponse> responseObserver) {

    }


    @Override
    public void receiveMessage(ReceiveMessageRequest request, StreamObserver<ReceiveMessageResponse> responseObserver) {

    }



    @Override
    public void ackMessage(AckMessageRequest request, StreamObserver<AckMessageResponse> responseObserver) {

    }


    @Override
    public void forwardMessageToDeadLetterQueue(ForwardMessageToDeadLetterQueueRequest request,
                                                StreamObserver<ForwardMessageToDeadLetterQueueResponse> responseObserver) {

    }


    @Override
    public void endTransaction(EndTransactionRequest request, StreamObserver<EndTransactionResponse> responseObserver) {

    }


    @Override
    public void notifyClientTermination(NotifyClientTerminationRequest request,
                                        StreamObserver<NotifyClientTerminationResponse> responseObserver) {

    }



    @Override
    public void changeInvisibleDuration(ChangeInvisibleDurationRequest request,
                                        StreamObserver<ChangeInvisibleDurationResponse> responseObserver) {

    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：处理客户端遥测请求的方法
     */
    @Override
    public StreamObserver<TelemetryCommand> telemetry(StreamObserver<TelemetryCommand> responseObserver) {
        Function<Status, TelemetryCommand> statusResponseCreator = status -> TelemetryCommand.newBuilder().setStatus(status).build();
        //这里的这个操作非常重要，这里返回了一个ContextStreamObserver对象，这个ContextStreamObserver对象的onNext方法中封装了真正处理客户端请求的操作
        ContextStreamObserver<TelemetryCommand> responseTelemetryCommand = grpcMessingActivity.telemetry(responseObserver);
        return new StreamObserver<TelemetryCommand>() {
            @Override
            public void onNext(TelemetryCommand value) {
                ProxyContext context = createContext();
                try {
                    validateContext(context);
                    addExecutor(clientManagerThreadPoolExecutor,
                            context,
                            value,
                            //在这里执行了真正处理客户端请求的操作
                            () -> responseTelemetryCommand.onNext(context, value),
                            responseObserver,
                            statusResponseCreator);
                } catch (Throwable t) {
                    writeResponse(context, value, null, responseObserver, t, statusResponseCreator);
                }
            }

            @Override
            public void onError(Throwable t) {
                responseTelemetryCommand.onError(t);
            }

            @Override
            public void onCompleted() {
                responseTelemetryCommand.onCompleted();
            }
        };
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/3
     * @方法描述：像执行器添加任务的方法，这里我要给大家解释一下，当Grpc服务端的GrpcMessagingApplication应用处理器执行客户端远程调用的目标方法时
     * 也就是处理客户端发送过来的请求时，会把请求提交给对应的线程池，这个方法就是用来向线程池提交处理请求的任务的
     */
    protected <V, T> void addExecutor(ExecutorService executor, ProxyContext context, V request, Runnable runnable,
                                      StreamObserver<T> responseObserver, Function<Status, T> statusResponseCreator) {
        //向线程池提交处理请求的任务，这里可以看到，runnable是真正处理请求的任务，但是在这里又对这个任务进行了一次封装，封装成了GrpcTask对象
        //这是因为当任务被线程池拒绝执行时，会调用GrpcTaskRejectedExecutionHandler的rejectedExecution方法，这个方法会把任务转换成GrpcTask对象
        //而GrpcTask对象中持有了任务被拒绝执行时要回复给客户端的响应码，所以这里对任务进行了一次封装
        //说得简单点，GrpcTask对象封装了当前任务的一些相关信息，比如请求对象，响应对象，执行任务的ProxyContext上下文信息等等
        //注意，这里执行了statusResponseCreator.apply(flowLimitStatus())这行代码会生成一个响应对象，这个响应对象的状态码是TOO_MANY_REQUESTS
        //这个响应对象会封装到GrpcTask对象中，当任务被线程池拒绝执行时，会使用这个响应对象回复给客户端，告诉客户端请求发送得太频繁了，需要限流
        executor.submit(new GrpcTask<>(runnable, context, request, responseObserver, statusResponseCreator.apply(flowLimitStatus())));
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/3
     * @方法描述：向客户端回复响应的方法
     */
    protected <V, T> void writeResponse(ProxyContext context, V request, T response, StreamObserver<T> responseObserver,
                                        Throwable t, Function<Status, T> errorResponseCreator) {
        //判断是否有异常信息
        if (t != null) {
            //如果有异常则调用errorResponseCreator.apply(convertExceptionToStatus(t)这行代码，convertExceptionToStatus(t)会根据异常信息生成一个status状态对象
            //而errorResponseCreator.apply()方法会根据这个状态对象生成一个响应对象，这个响应对象就是错误响应对象
            //再使用responseObserver对象把把响应回复给客户端即可
            ResponseWriter.getInstance().write(responseObserver, errorResponseCreator.apply(convertExceptionToStatus(t)));
        } else {
            //没有异常信息则直接回复客户端响应
            ResponseWriter.getInstance().write(responseObserver, response);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/3
     * @方法描述：创建请求上下文信息的方法
     */
    protected ProxyContext createContext() {
        //从LocalThread中获取当前现成的私有上下文信息
        Context ctx = Context.current();
        //从私有上下文信息中获取请求元数据信息
        Metadata headers = InterceptorConstants.METADATA.get(ctx);
        //创建ProxyContext对象，使用元数据信息填充该ProxyContext对象的属性
        ProxyContext context = ProxyContext.create()
                //设置本地地址
                .setLocalAddress(getDefaultStringMetadataInfo(headers, InterceptorConstants.LOCAL_ADDRESS))
                //设置远程地址
                .setRemoteAddress(getDefaultStringMetadataInfo(headers, InterceptorConstants.REMOTE_ADDRESS))
                //设置客户端ID
                .setClientID(getDefaultStringMetadataInfo(headers, InterceptorConstants.CLIENT_ID))
                //设置协议类型
                .setProtocolType(ChannelProtocolType.GRPC_V2.getName())
                //设置编程语言
                .setLanguage(getDefaultStringMetadataInfo(headers, InterceptorConstants.LANGUAGE))
                //设置客户端版本
                .setClientVersion(getDefaultStringMetadataInfo(headers, InterceptorConstants.CLIENT_VERSION))
                //设置操作名称
                .setAction(getDefaultStringMetadataInfo(headers, InterceptorConstants.SIMPLE_RPC_NAME))
                //设置命名空间
                .setNamespace(getDefaultStringMetadataInfo(headers, InterceptorConstants.NAMESPACE_ID));
        //设置请求完成时间，在第四版本代码中这个请求完成时间根本用不到，等后面实现了客户端，并且实现了消费者客户端的receiveMessage()方法后，这个属性才会用到
        if (ctx.getDeadline() != null) {
            //设置请求完成的截止时间
            context.setRemainingMs(ctx.getDeadline().timeRemaining(TimeUnit.MILLISECONDS));
        }
        return context;
    }



    //校验上下文信息是否合规的方法
    protected void validateContext(ProxyContext context) {
        if (StringUtils.isBlank(context.getClientID())) {
            throw new GrpcProxyException(Code.CLIENT_ID_REQUIRED, "client id cannot be empty");
        }
    }

    //从元数据信息中获取对应的字符串信息的方法
    protected String getDefaultStringMetadataInfo(Metadata headers, Metadata.Key<String> key) {
        return StringUtils.defaultString(headers.get(key));
    }


    //终止当前服务组件工作的方法
    @Override
    public void shutdown() throws Exception {
        this.grpcMessingActivity.shutdown();
        this.routeThreadPoolExecutor.shutdown();
        this.clientManagerThreadPoolExecutor.shutdown();
    }

    //启动当前服务组件工作的方法
    @Override
    public void start() throws Exception {
        this.grpcMessingActivity.start();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/3
     * @方法描述：这个内部类的对象就是要被提交给线程池的Runnable任务
     */
    protected static class GrpcTask<V, T> implements Runnable {
        //真正处理请求的任务
        protected final Runnable runnable;
        //请求上下文信息对象
        protected final ProxyContext context;
        //请求对象
        protected final V request;
        //任务被线程池拒绝执行时要回复给客户端的响应
        protected final T executeRejectResponse;
        //用来回复客户端响应的对象
        protected final StreamObserver<T> streamObserver;


        public GrpcTask(Runnable runnable, ProxyContext context, V request, StreamObserver<T> streamObserver,
                        T executeRejectResponse) {
            this.runnable = runnable;
            this.context = context;
            this.streamObserver = streamObserver;
            this.request = request;
            this.executeRejectResponse = executeRejectResponse;
        }


        @Override
        public void run() {
            this.runnable.run();
        }
    }


    //线程池拒绝处理器
    protected class GrpcTaskRejectedExecutionHandler implements RejectedExecutionHandler {

        public GrpcTaskRejectedExecutionHandler() {

        }

        //当线程池中的任务被拒绝执行时，该方法就会被调用
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            if (r instanceof GrpcTask) {
                try {
                    GrpcTask grpcTask = (GrpcTask) r;
                    //在这里把默认的响应对象回复给客户端了
                    writeResponse(grpcTask.context, grpcTask.request, grpcTask.executeRejectResponse, grpcTask.streamObserver, null, null);
                } catch (Throwable t) {
                    log.warn("write rejected error response failed", t);
                }
            }
        }
    }

}
